package com.chukun.flink.stream.window.process;

import com.chukun.flink.stream.window.source.SourceForWindow;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author chukun
 * @version 1.0.0
 * @description 窗口 process 函数操作
 * @createTime 2022年05月22日 23:38:00
 */
public class ProcessWindowOperator {

    public static void main(String[] args) throws Exception {

        // 创建运行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置作业的全局并行度 1
        env.setParallelism(1);

        // 添加数据源
        DataStreamSource<Tuple3<String, Integer, String>> stremSource = env.addSource(new SourceForWindow(1000, false));

        // 根据数据流中的元素f0字段作为作为key对数据流分组
        DataStream<String> reduceStream = stremSource.keyBy((key) -> key.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // 对窗口应用ProcessFunction窗口函数
                .process(new ProcessWindowFunction<Tuple3<String, Integer, String>, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, ProcessWindowFunction<Tuple3<String, Integer, String>, String, String, TimeWindow>.Context ctx, Iterable<Tuple3<String, Integer, String>> input, Collector<String> collector) throws Exception {
                        String str = "";
                        long count = 0;
                        for (Tuple3<String, Integer, String> in : input) {
                            str = StringUtils.join(str, in.toString());
                            count++;
                        }
                        System.out.printf("process窗口元素:%s", str);
                        collector.collect("Window : "+ctx.window() +" key: "+key+" count: "+count);
                    }
                });

        reduceStream.print("窗口-process计算: ");

        env.execute("ProcessWindowOperator");
    }
}
